package com.atguigu.flink.chapter08_exec2;

import com.atguigu.flink.pojo.AdsClickLog;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;


/**
 * Created by Smexy on 2022/10/29
 *
 *  以天为单位，统计每个用户对广告的点击量。
 *      如果同一个广告，同一个用户，一天内点击超过100次，拉黑预警。
 *
 *  --------------------------------------
 *
 *      采取状态来解决。
 *          如何判断当前已经跨天了呢？ 通过状态来标记。
 *              同一个用户在不同日期点击的数据，每次处理时，记录下当前的日期，假设下一条日期变了，认为跨天了。
 *              前提： 数据必须是有序的。
 *
 *
 *      不开窗，无需水印。水印是基于事件时间的窗口，计算时才需要！
 *              定时器需要！
 *
 *       保存的状态:
 *          1.日期 string :  ValueState
 *          2.是否被拉黑了，如果已经拉黑，他拉黑后续的点击记，没必要统计
 *          3.点击量聚合统计，才知道是否应该被拉黑。
 *
 *    ----------------------
 *      map的并行度也需要设置为1
 *
 *
 */
public class Demo4_AdsClick
{
    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.readTextFile("data/AdClickLog.csv").setParallelism(1)
           .map(new MapFunction<String, AdsClickLog>()
           {
               @Override
               public AdsClickLog map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new AdsClickLog(
                       Long.parseLong(words[0]),
                       Long.parseLong(words[1]),
                       words[2],
                       words[3],
                       Long.parseLong(words[4]) * 1000
                   );
               }
           })
           //统计的一个用户对一种广告的点击
           .keyBy(log -> log.getUserId() + "_" + log.getAdsId())
           .process(new KeyedProcessFunction<String, AdsClickLog, String>()
           {

               private ValueState<Boolean> ifBlack;
               private ValueState<Integer> click;
               private ValueState<String> currentDate;

               @Override
               public void open(Configuration parameters) throws Exception {
                   RuntimeContext context = getRuntimeContext();

                   currentDate = context.getState(new ValueStateDescriptor<String>("currentDate", String.class));
                   click = context.getState(new ValueStateDescriptor<Integer>("click", Integer.class));
                   ifBlack = context.getState(new ValueStateDescriptor<Boolean>("ifBlack", Boolean.class));


               }

               //用不用做成状态?  不用
               // 什么变量需要做成状态？  共用(每一条数据都要用)的写的变量才需要制作为状态
               //  共用的，只读变量，无需作为状态
               SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
               @Override
               public void processElement(AdsClickLog value, Context ctx, Collector<String> out) throws Exception {

                   //当前数据携带的日期
                   String day = this.format.format(new Date(value.getTimestamp()));

                   if (!day.equals(currentDate.value())){
                       //跨天重置状态
                       currentDate.update(day);
                       System.out.println(value.getUserId() + "点击 " + value.getAdsId() +" 日期重置为:"+day);
                       ifBlack.clear();
                       click.clear();
                   }

                   //统计点击
                   if (click.value() == null){
                       click.update(1);
                   }else{
                       click.update(click.value() + 1);
                   }

                   //输出信息
                   String msg = value.getUserId() + " 点击了 "+value.getAdsId() + " 次数: " + click.value();

                   //根据点击量判断是否需要拉黑
                   if (click.value() > 100) {
                       //在第一次点击次数超过 100的时候，被触发一次就行了
                       if (ifBlack.value() == null){
                           ifBlack.update(true);
                           out.collect(msg + ",已拉黑...");
                       }
                   }else{
                       //正常
                       out.collect(msg);
                   }

               }
           }).print().setParallelism(1);


        try {
                    env.execute();
                } catch (Exception e) {
                    e.printStackTrace();
                }
    }
}
